#include "aeEvents.hpp"
#include "network.hpp"
#include "Util.hpp"
#include "HashObject.hpp"
#include "listObject.hpp"
#include "stringObject.hpp"
#include "set.hpp"
#include "ZsetObject.hpp"

// AOF刷新内核级缓冲区的频率
#define APPENDSYNC_ALWAYS 1   // 每次都刷新
#define APPENDSYNC_EVERYSEC 2 // 每隔一秒刷新
#define APPENDSYNC_NO 3       // 不刷新，刷新时机交给操作系统

// client flag
#define LOADINGAOF 1 // 伪客户端，正在持久化
#define MUILTI 2     // 事务开始
#define DIRTY_DAS 4  // 事务安全性遭到破坏不能执行此事务

// redis state
#define BGREWRITEAOF 1

using namespace redisStruct;

struct redisClient;
struct redisCommond;

static void AcceptHandler(aeEvents *aeEvt, int sockfd, void *pridata);
static void call(redisClient *client, redisCommond *cmd);
static void feedReWriteAOFBuffer(vector<string> &args);
static void feedAppendOnlyFile(int db_id, vector<string> &args);
static void loadAppendOnlyFile(std::string filename);
static int ServerCron();
static void createClient(int fd);
static void closeClient(redisClient *client);
static void handlerClientInbuf(redisClient *client);
static void TryToSendClientOutBuf(redisClient *client);
static void readTcpStreamToClientInbuf(aeEvents *aeEv, int fd, void *pridata);
static void sendClientOutbufToTcpStream(aeEvents *aeEvt, int fd, void *pridata);
static int selectDb(redisClient *client, int db_id);
static void NotifyClientWatchedKeyChanged(int db_id, stringObject *key);
static void bgrewriteaof();
static void handlerChLDSignal(int sig);


/**********针对字符串对象类型的命令****/
static void getCommond(redisClient *);
static void setCommond(redisClient *);
static void setnxCommond(redisClient *);
static void appendCommond(redisClient *);
static void substrCommond(redisClient *);
static void incrCommond(redisClient *);
static void decrCommond(redisClient *);
static void mgetCommond(redisClient *);

/**********针对链表对象的命令**********/
static void rpushCommond(redisClient *);
static void lpushCommond(redisClient *);
static void rpopCommond(redisClient *);
static void lpopCommond(redisClient *);
static void lrangeCommond(redisClient *);
static void llenCommond(redisClient *);
static void lsetCommond(redisClient *);
static void ltrimCommond(redisClient *);
static void lremCommond(redisClient *);
static void rpoplpushCommond(redisClient *);

/********针对哈希对象的命令**********/
static void hsetCommond(redisClient *);
static void hsetnxCommond(redisClient *);
static void hgetallCommond(redisClient *);
static void hgetCommond(redisClient *);
static void hdelCommond(redisClient *);
static void hlenCommond(redisClient *);
static void hkeysCommond(redisClient *);
static void hvalsCommond(redisClient *);

/********针对集合对象的命令**********/
static void saddCommond(redisClient *);
static void smembersCommond(redisClient *);
static void scardCommond(redisClient *);
static void sremCommond(redisClient *);
static void smoveCommond(redisClient *);
static void sinterCommond(redisClient *);
static void sunionCommond(redisClient *);
static void sdiffCommond(redisClient *);

/**********针对有序集合对象的命令*********/
static void zaddCommond(redisClient *);
static void zrangeCommond(redisClient *);
static void zcountCommond(redisClient *);
static void zincrbyCommond(redisClient *);
static void zscoreCommond(redisClient *);
static void zrangeByscoreCommond(redisClient *);
static void zrankCommond(redisClient *);
static void zremCommond(redisClient *);

/********通用命令*****************/
static void objectEcondingCommond(redisClient *);
static void delCommond(redisClient *);
static void existsCommond(redisClient *);
static void selectCommond(redisClient *);
static void multiCommond(redisClient *);
static void execCommond(redisClient *);
static void watchCommond(redisClient *);
static void pexpireatCommond(redisClient *);
static void subscribeCommond(redisClient *);
static void unsubscribeCommond(redisClient *);
static void publishCommond(redisClient* );
static void psubscribeCommond(redisClient* );
static void punsubscribeCommond(redisClient *);
static void bgrewriteaofCommond(redisClient*);

struct sharedStrings
{
    string OK = "OK\r\n";
    string CRLF = "\r\n";
    string NotExist = "Key not existed\r\n";
    string UnknownCmd = "Unkouwn commond\r\n";
    string TypeMisMatch = "Type MisMatch\r\n";
    string ArgumentNumError = "The number of parameters is incorrect\r\n";
} shared;

typedef void commondProc(redisClient *);
struct redisCommond
{
    string name;
    int args;
    bool dirty;
    commondProc *proc;
};
redisCommond cmds[] = {

    /*针对字符串对象的命令*/
    {"get", 2, false, getCommond},       // 获取key对应字符串
    {"set", 3, true, setCommond},        // 若key已存在则覆盖
    {"setnx", 3, true, setnxCommond},    // 若key已存在则失败
    {"append", 3, true, appendCommond},  // 给字符串添加字符串，若key不存在则相当于set
    {"substr", 4, false, substrCommond}, // 求子串[i,j] j负数时表示倒数第j位 ： [1,-2]：表示第1位 到 倒数第1位
    {"incr", 2, true, incrCommond},      // +1
    {"decr", 2, true, decrCommond},      // -1
    {"mget", -2, true, mgetCommond},     // 获取key对应字符串，支持多个key

    /*针对链表对象的命令*/
    {"rpush", -3, true, rpushCommond},   // 尾插
    {"lpush", -3, true, lpushCommond},   // 头插
    {"rpop", 2, true, rpopCommond},      // 尾删
    {"lpop", 2, true, lpopCommond},      // 头删
    {"lrange", 4, false, lrangeCommond}, // 获取链表【start,end】
    // blpop brpop
    {"llen", 2, false, llenCommond},          // 获取链表长度
    {"lset", 4, true, lsetCommond},           // 通过下标设置元素值
    {"ltrim", 4, true, ltrimCommond},         // 保留区域内元素
    {"lrem", 3, true, lremCommond},           // 删除相同元素
    {"rpoplpush", 2, true, rpoplpushCommond}, // 将尾部元素与弹出 移到首部

    /*针对哈希表对象的命令*/
    {"hset", -4, true, hsetCommond},       // 添加键值对，键已存在则修改值
    {"hsetnx", -4, true, hsetnxCommond},   // 如果键已存在则不修改
    {"hgetall", 2, false, hgetallCommond}, // 获取哈希所有键值对
    {"hget", 3, false, hgetCommond},       // 获取key对应的val
    {"hdel", 3, true, hdelCommond},        // 删除key
    {"hlen", 2, false, hlenCommond},       // 获取键值对数
    {"hkeys", 2, false, hkeysCommond},     // 获取所有k
    {"hvals", 2, false, hvalsCommond},     // 获取所有v

    /*针对集合类型的对象*/
    {"sadd", -3, true, saddCommond},         // 集合添加元素，元素已存在则忽略
    {"smembers", 2, false, smembersCommond}, // 获取集合所有元素
    {"scard", 2, false, scardCommond},       // 集合元素个数
    {"srem", -3, true, sremCommond},         // 移除集合一个或多个元素
    {"smove", 4, true, smoveCommond},        // 将元素从x集合移到y集合
    {"sinter", -2, false, sinterCommond},    // 求交集
    {"sunion", -2, false, sunionCommond},    // 求并集
    {"sdiff", -2, false, sdiffCommond},      // 求第一个集合与其他集合的差集

    /*针对有序集合类型的对象*/
    {"zadd", -4, true, zaddCommond},                   // 插入元素
    {"zrange", 4, false, zrangeCommond},               // 返回区间内的元素 , 通过下标
    {"zrangebyscore", 4, false, zrangeByscoreCommond}, // 返回区间内的元素 , 通过分数
    {"zcount", 4, false, zcountCommond},               // 返回分数在【min,max】成员数
    {"zincrby", 4, true, zincrbyCommond},              // 对指定成员的分数加上增量 increment
    {"zscore", 3, false, zscoreCommond},               // 返回对应成员的分数
    {"zrank", 3, false, zrankCommond},                 // 返回成员的索引
    {"zrem", -3, true, zremCommond},                   // 移除一个或多个成员

    /*通用命令*/
    {"objectencoding", 2, false, objectEcondingCommond}, //*****TODO*********求对象底层编码
    {"del", 2, true, delCommond},                        // 删除key
    {"exists", 2, false, existsCommond},                 // 判断key是否存在
    {"select", 2, false, selectCommond},                 // 选择数据库
    {"multi", 1, false, multiCommond},                   // 开启事务
    {"exec", 1, false, execCommond},                     // 执行事务
    {"watch", 2, true, watchCommond},                    // 监视key, watch命令虽然不修改数据但会影响事务运行，因此需要被持久化到文件里
    {"pexpireat", 3, true, pexpireatCommond},            // 给key设置过期时间，精度为毫秒   *****有关的命令TODO
    {"subscribe",2,false,subscribeCommond},              // 订阅频道
    {"unsubscribe",2,false,unsubscribeCommond},          // 退订频道
    {"publish",3,false,publishCommond},                  // 发布消息
    {"psubscribe",2,false,psubscribeCommond},            // 订阅模式 
    {"punsubscribe",2,false,punsubscribeCommond},        // 退订模式  
    {"bgrewriteaof",1,false,bgrewriteaofCommond}         // 重写AOF文件

};
struct redisDb
{
    // int _id = -1;
    unordered_map<stringObject *, redisObject *> _dict;
    unordered_map<stringObject *, std::list<redisClient *>> _watchkeys;
    unordered_map<stringObject *, long long> _expire; // 存键的过期时间，精度为毫秒
};
struct multiCmd
{
    std::vector<string> args;
    redisCommond *cmd;
};
struct redisClient
{
    int _fd = -1;
    int _db_id = 0; // 客户端当前所选数据库下标 合法范围[0~15]
    int flag = 0;
    string _inbuf;
    std::list<string> _outbuf;
    vector<string> args;
    std::list<multiCmd> multiQueue;
    vector<stringObject *> _watchkeys;
    redisClient(int fd = -1) : _fd(fd)
    {
    }
    void addReply(const string &reply)
    {
        if (flag & LOADINGAOF)
            return;

        _outbuf.push_back(reply);
        TryToSendClientOutBuf(this);
    }
    ~redisClient()
    {
    }
};
struct pubsub_pattern
{
   std::string pattern; //频道
   redisClient* client;
};
struct RedisServer
{
    int _fd;
    int _port = 8080;
    redisDb _db[16];
    aeEvents _aeEv;
    std::list<redisClient *> _clients;
    std::unordered_map<std::string,std::list<redisClient*>> _pubsub_channels;//订阅频道
    std::list<pubsub_pattern> pubsub_patterns;//订阅模式
    int _aof_selectDb;
    std::string _aofname = "aof.data";
    int _aof_fd;
    time_t unxitime;
    int appendfsync = APPENDSYNC_EVERYSEC;
    time_t lastfsync = 0;
    int state=0;
    pid_t bgrewriteaofPid=-1;
    std::string rewriteaofBuffer;
    RedisServer()
    {
        initServer();
    }
    ~RedisServer()
    {
        close(_fd);
        close(_aof_fd);
        while (!_clients.empty())
        {
            redisClient *client = _clients.front();
            close(client->_fd);
            _clients.pop_front();
        }
    }
    void initServer()
    {
        // 建立套接字，并注册相应事件
        _fd = network::TcpServer(_port);
        setFdNoBlock(_fd);
        _aeEv.registeFileEvent(_fd, AE_READABLE, nullptr, AcceptHandler, nullptr);

        long sec, ms;
        aeGetTime(sec, ms);
        aeAddTime(sec, ms, 500);
        _aeEv.registerTimeEvent(sec, ms, ServerCron);

        // 注册信号处理函数
        if(signal(SIGCHLD,handlerChLDSignal) == SIG_ERR)
        {
            cout<<"注册信号处理函数失败!"<<endl;
            exit(-1);
        }

        // 加载aof文件
        loadAppendOnlyFile(_aofname);
        
        // 打开aof文件，进行持久化
        umask(0);
        _aof_fd = open(_aofname.c_str(), O_CREAT | O_APPEND | O_WRONLY, 0777);
        if (_aof_fd < 0)
        {
            cout << "open aof_file:" << _aofname << " fail!" << endl;
            exit(-1);
        }

        cout << "open aof_file:" << _aofname << endl;
        cout << "start aof function" << endl;
    }

    void addClient(redisClient *client)
    {
        _clients.push_back(client);
    }
    void removeClient(redisClient *client)
    {
        _clients.remove(client);
    }
} redis_server;

static void call(redisClient *client, redisCommond *cmd)
{
    cmd->proc(client);

    // 持久化
    if (cmd->dirty == true && !(client->flag & LOADINGAOF))
    {
        if(redis_server.state&BGREWRITEAOF)
        feedReWriteAOFBuffer(client->args);//将命令添加到aof重写缓冲区
        else
        feedAppendOnlyFile(client->_db_id, client->args);//将命令添加到aof文件

    }
}
static void feedReWriteAOFBuffer(vector<string> &args)
{
    cout<<"feedReWriteAOFBuffer"<<endl;
    
    for(auto& str : args)
        redis_server.rewriteaofBuffer+=str+" ";

    redis_server.rewriteaofBuffer.resize(redis_server.rewriteaofBuffer.size()-1);   
    redis_server.rewriteaofBuffer+="\r\n";
}

static void feedAppendOnlyFile(int db_id, vector<string> &args)
{
    std::string str;

    // 如果服务器默认选择的数据库与客户端的数据库不一致则更改服务器选择的数据库
    if (db_id != redis_server._aof_selectDb)
    {
        str = "select " + std::to_string(db_id) + shared.CRLF;
        redis_server._aof_selectDb = db_id;
    }

    // 添加命令
    for (int i = 0; i < args.size(); i++)
    {
        if (i == args.size() - 1)
        {
            str += args[i] + shared.CRLF;
        }
        else
        {
            str += args[i] + " ";
        }
    }

    int n = write(redis_server._aof_fd, str.c_str(), str.size());
    if (n != str.size())
    {
        if (n == -1)
            cerr << "feed aof error! aof_fd:" << redis_server._aof_fd << endl;
        else
            cerr << "aof write too short! aof_fd:" << redis_server._aof_fd << endl;
        exit(-1);
    }

    // 根据刷新策略 刷新内核级缓冲区
    if (redis_server.appendfsync == APPENDSYNC_ALWAYS)
        fsync(redis_server._aof_fd);
    else if (redis_server.appendfsync == APPENDSYNC_EVERYSEC)
    {
        time_t now = time(nullptr);
        if (now - redis_server.lastfsync >= 50)
        {
            fsync(redis_server._aof_fd);
            redis_server.lastfsync = now;
        }
    }
}

static void loadAppendOnlyFile(std::string filename)
{
    cout << "start loading AppendOnlyFile" << endl;
    int fd = open(filename.c_str(), O_RDONLY);
    if (fd < 0)
        cerr << "load aof fail!" << endl;

    redisClient fakeClient;
    fakeClient.flag |= LOADINGAOF;
    int n;
    while (1)
    {
        n = readLine(fd, fakeClient._inbuf);
        // cout<<fakeClient._inbuf;
        if (n <= 0)
            break;
        handlerClientInbuf(&fakeClient);
    }
    if (n == 0)
        close(fd);
    else
    { // 读出错
        close(fd);
        exit(-1);
    }
    cout << "loaded AppendOnlyFile success" << endl;
}

static int ServerCron()
{
    redis_server.unxitime = time(nullptr);

    // cout<< "ServerCron()" <<endl;
    return 500; // 周期事件，每500毫秒执行一次
}
static void createClient(int fd)
{
    setFdNoBlock(fd);
    redisClient *client = new redisClient(fd);
    redis_server.addClient(client);
    // cout<<"create fd"<<fd<<endl;
    redis_server._aeEv.registeFileEvent(fd, AE_READABLE, client, readTcpStreamToClientInbuf, nullptr);
    cout << "create a client" << endl;
}

static void closeClient(redisClient *client)
{
    close(client->_fd);
    redis_server.removeClient(client);
    redis_server._aeEv.deleteFileEvent(client->_fd);

    cout << "close a client(fd:" << client->_fd << ")" << endl;
    delete client;
}

static void readTcpStreamToClientInbuf(aeEvents *aeEv, int fd, void *pridata)
{
    // REDIS_NOTUSED(aeEv);
    int nums = aeRead(fd, ((redisClient *)pridata)->_inbuf);

    if (nums <= 0)
    {
        closeClient((redisClient *)pridata);
        return;
    }

    handlerClientInbuf((redisClient *)pridata);
}

static void handlerClientInbuf(redisClient *client)
{ // 根据空格分隔参数
    std::stringstream ss(client->_inbuf);
    string str;
    while (ss >> str)
    {
        client->args.push_back(str);
    }
    if (client->args.size() == 0)
        return;

    // 查找命令
    int cmdnums = sizeof(cmds) / sizeof(redisCommond);
    redisCommond *cmd = nullptr;
    std::transform(client->args[0].begin(), client->args[0].end(), client->args[0].begin(), tolower);

    for (int i = 0; i < cmdnums; i++)
    {

        // cout<<cmds[0].name<<endl;
        if (client->args[0] == cmds[i].name)
        {
            cmd = cmds + i;
            break;
        }
    }

    if (cmd == nullptr)
    {
        // aeWrite(client->_fd, shared.UnknownCmd);
        client->addReply(shared.UnknownCmd);
    }
    else if ((cmd->args > 0 && cmd->args != client->args.size()) || cmd->args < 0 && (-1 * cmd->args) > client->args.size())
    {
        // aeWrite(client->_fd, cmd->name + string(" arguments should be ") + std::to_string(cmd->args) + sharedReply.LF);
        client->addReply(shared.ArgumentNumError);
    }
    else
    {
        // 参数正确

        // 处于事务中并且命令不是exec,入命令队列
        if (client->flag & MUILTI && client->args[0] != "exec")
        {
            client->multiQueue.push_back({client->args, cmd});
            client->addReply("Queued\r\n");
        }
        else
            call(client, cmd);
    }

    client->_inbuf.clear();
    client->args.clear();
    return;
}

// 循环拿链接，并创建客户端，注册事件
void AcceptHandler(aeEvents *aeEvt, int sockfd, void *pridata)
{
    REDIS_NOTUSED(pridata);
    while (1)
    {
        int fd = network::TcpAccept(sockfd);
        if (fd >= 0)
        {
            cout << "accept a link(fd:" << fd << ")" << endl;
            createClient(fd);
        }
        else if (fd == -1)
        {
            if (errno == EINTR)
                continue;
            else if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
                return;
            }
            else
            { // 套接字读出错
                cerr << "AcceptHandler error, fd:" << sockfd << endl;
                return;
            }
        }
    }
}

static void TryToSendClientOutBuf(redisClient *client)
{
    if (client->_outbuf.size() == 0)
        return;

    if (!redis_server._aeEv.IsVolidFd(client->_fd))
        return;

    if (redis_server._aeEv.getFdEvents(client->_fd) & AE_WRITEABLE)
        return;

    redis_server._aeEv.addFileEvent(client->_fd, AE_WRITEABLE, nullptr, sendClientOutbufToTcpStream);
}

static void sendClientOutbufToTcpStream(aeEvents *aeEvt, int fd, void *pridata)
{
    redisClient *client = (redisClient *)pridata;

    while (client->_outbuf.size() > 0)
    {
        if (aeWrite(fd, client->_outbuf.front()) < 0)
        {
            closeClient(client);
            break;
        }
        client->_outbuf.pop_front();
    }

    if (client->_outbuf.size() == 0)
        redis_server._aeEv.deleteFileEvent(fd, AE_WRITEABLE);
}
static int selectDb(redisClient *client, int db_id)
{
    if (db_id < 0 || db_id > 16)
        return REDIS_ERROR;

    client->_db_id = db_id;

    return REDIS_OK;
}

static void NotifyClientWatchedKeyChanged(int db_id, stringObject *key)
{
    auto &watchkey = redis_server._db->_watchkeys;
    auto ret = watchkey.find(key);
    if (ret != watchkey.end())
    {
        for (auto &e : ret->second)
            e->flag |= DIRTY_DAS;
    }
}

static void bgrewriteaof()
{
    cout<<"bgrewriteaof"<<endl;
    umask(0);
    int fd = open("rwaof.data",O_WRONLY|O_CREAT,0777);

    if(fd<0)
    {
        exit(-1);
        cerr<<"rwaof open fail!\r\n";
    }
    
    std::string buf;
    for(int i=0;i<16;i++)
    {
        if(redis_server._db[i]._dict.size() == 0 ) continue;
        
        buf+="select "+std::to_string(i)+"\r\n";
        
        for(auto& pair : redis_server._db[i]._dict)
        {
            redisObject*  redisobj = pair.second;
            if(redisobj->_type == TYPE_STRING)
            {
                buf+="set "+string(pair.first->c_str(),pair.first->_size)+" ";
                stringObject* strobj = (stringObject*)redisobj;
                if(strobj->_encoding == TYPE_Encoding_INT)
                buf+=std::to_string((long)strobj->c_str())+"\r\n";   
                else if(strobj->_encoding == TYPE_Encoding_SDS)
                buf+=string(strobj->c_str(),strobj->_size)+"\r\n";          
            }
            else if(redisobj->_type == TYPE_LIST)
            {
                listObject* listobj = (listObject*)redisobj;
                buf+="rpush "+string(pair.first->c_str(),pair.first->_size)+" ";

                for(auto& e : *listobj)
                {
                    if(e->_encoding==TYPE_Encoding_SDS)
                    buf+=string(e->c_str(),e->_size)+" ";
                    else if(e->_encoding == TYPE_Encoding_INT)
                    buf+=std::to_string((long)e->c_str())+" ";
                }
                
                buf.resize(buf.size()-1);
                buf+="\r\n";
            }
            else if(redisobj->_type == TYPE_HASH)
            {
                auto hashobj = (HashObject*)redisobj;
                buf+="hset "+string(pair.first->c_str(),pair.first->_size)+" ";
                for(auto& pair : *hashobj)
                {
                    buf+=std::string(pair.first->c_str(),pair.first->_size)+" ";
                    buf+=std::string(pair.second->c_str(),pair.second->_size)+" ";
                }
                buf.resize(buf.size()-1);
                buf+="\r\n";
            }
            else if(redisobj->_type == TYPE_SET)
            {
                buf+="sadd "+string(pair.first->c_str(),pair.first->_size)+" ";
                SetObject* setobj = (SetObject*)redisobj;
                if(setobj->_encoding == TYPE_Ecoding_HASHTABLE)
                {
                    auto* hashtable = (unordered_set<stringObject*>*)setobj->_ptr;
                    for(auto* str : *hashtable)
                    {
                        if(str->_encoding == TYPE_Encoding_SDS)
                        buf+=string(str->c_str(),str->_size)+" ";
                        else if(str->_encoding == TYPE_Encoding_INT)
                        buf+=string(std::to_string((long)str->c_str())+" ");
                    }
                }
                else if(setobj->_encoding == TYPE_Ecoding_INTSET)
                {
                   intset* pintset = (intset*)setobj->_ptr;
                   for(int i=0;i<pintset->_length;i++)
                   buf += string(std::to_string(intset::getSizePtr(pintset->_data+i*setobj->_encoding,setobj->_encoding))) + " ";
                }
                buf.resize(buf.size()-1);
                buf+="\r\n";
            }
            else if(redisobj->_type == TYPE_ZSET)
            {
                buf+="zadd "+string(pair.first->c_str(),pair.first->_size)+" ";
                ZsetObject* zsetobj = (ZsetObject*)redisobj;
                auto con = zsetobj->zrange(0,zsetobj->size()-1);
                for(auto& pair : con)
                {
                   buf+=std::to_string(pair.first)+" ";
                   buf+=string(pair.second->c_str(),pair.second->_size)+" ";
                }
                buf.resize(buf.size()-1);
                buf+="\r\n";
            }

            int ret = write(fd,buf.c_str(),buf.size());
            buf.clear();
            if(ret<0)
            {
                exit(-1);
                cerr<<"rewrite fail!\r\n";
            }
      
        }
    }

    close(fd);
}

static void handlerChLDSignal(int sig)
{
   cout<<"handlerChLDSignal"<<endl;
   int ret;
   int r = waitpid(redis_server.bgrewriteaofPid,&ret,0);
   if(r!=-1)
   cout<<"rewrite sucess"<<endl;
   else
   cout<<"rewrite fail!"<<endl;
   
   close(redis_server._aof_fd);
   rename("rwaof.data","aof.data");
   int fd = open("aof.data",O_WRONLY|O_APPEND);
   if(fd<0)
   {
     cout<<"open rwaof.data fail!\r\n";  
     exit(-1);   
   }
   redis_server._aof_fd = fd;

   if(!redis_server.rewriteaofBuffer.empty())
   write(fd,redis_server.rewriteaofBuffer.c_str(),redis_server.rewriteaofBuffer.size());
   redis_server.rewriteaofBuffer.clear();
   
   redis_server.state&=~BGREWRITEAOF;
}

static void getCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *str1 = new stringObject(client->args[1].c_str(), client->args[1].size());
    str1->incrRefCount();

    auto ret = dict.find(str1);
    str1->decrRefCount();

    if (ret == dict.end())
        client->addReply(shared.NotExist);
    else
    {
        // 检查是否过期
        auto &expire = redis_server._db->_expire;
        auto iter = expire.find(str1);
        if (iter != expire.end())
        {
            if (aeGetTime() >= iter->second)
            { // 已经过期
                dict.erase(str1);
                expire.erase(str1);
                client->addReply(shared.NotExist);
                return;
            }
        }

        if (ret->second->_type != TYPE_STRING)
        {
            client->addReply(shared.TypeMisMatch);
        }
        else
        {
            stringObject *v = (stringObject *)ret->second;
            if (v->_encoding == TYPE_Encoding_SDS)
                client->addReply(string(v->c_str(), v->_size) + shared.CRLF);
            else
                client->addReply(std::to_string((long long)v->c_str()) + shared.CRLF);
        }
    }
}

static void setCommond(redisClient *client)
{
    stringObject *str1 = new stringObject(client->args[1].c_str(), client->args[1].size());
    str1->incrRefCount();
    stringObject *str2 = new stringObject(client->args[2].c_str(), client->args[2].size());
    str2->incrRefCount();

    redis_server._db[client->_db_id]._dict.replace(make_pair(str1, str2));

    // 检查key是否被监视
    /*auto& watchkey = redis_server._db->_watchkeys;
    auto ret = watchkey.find(str1);
    if(ret!=watchkey.end())
    {
       for(auto& e : ret->second)
       e->flag|= DIRTY_DAS;
    }*/
    NotifyClientWatchedKeyChanged(client->_db_id, str1);

    str1->decrRefCount();
    str2->decrRefCount();

    client->addReply(shared.OK);
}

static void setnxCommond(redisClient *client)
{
    stringObject *str1 = new stringObject(client->args[1].c_str(), client->args[1].size());
    str1->incrRefCount();
    stringObject *str2 = new stringObject(client->args[2].c_str(), client->args[2].size());
    str2->incrRefCount();

    auto rsl = redis_server._db[client->_db_id]._dict.insert(make_pair(str1, str2));

    str1->decrRefCount();
    str2->decrRefCount();
    client->addReply(shared.OK);

    if (rsl.second == false)
        return;

    // 检查key是否被监视
    /*auto& watchkey = redis_server._db->_watchkeys;
    auto ret = watchkey.find(str1);
    if(ret!=watchkey.end())
    {
       for(auto& e : ret->second)
       e->flag|= DIRTY_DAS;
    }*/
    NotifyClientWatchedKeyChanged(client->_fd, str1);
}

static void appendCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret == dict.end()) // key不存在
    {
        client->addReply(shared.NotExist);
    }
    else
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_STRING)
        {
            ((stringObject *)ret->second)->append(client->args[2].c_str(), client->args[2].size());
            client->addReply(shared.OK);
            NotifyClientWatchedKeyChanged(client->_db_id, key);
        }
        else // val的类型不是 string
            client->addReply(shared.TypeMisMatch);
    }

    key->decrRefCount();
}

static void substrCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    // 该key不存在
    if (ret == dict.end())
    {
        client->addReply(shared.NotExist);
    }
    else
    {
        if (ret->second->_type != TYPE_STRING)
        {
            client->addReply(shared.TypeMisMatch);
        }
        else
        {
            stringObject *val = (stringObject *)ret->second;

            char *ptr1 = nullptr;
            char *ptr2 = nullptr;
            int start_i = strtol(client->args[2].c_str(), &ptr1, 10);
            int len = strtol(client->args[3].c_str(), &ptr2, 10);

            if (ptr1 - client->args[2].c_str() == client->args[2].size() && ptr2 - client->args[3].c_str() == client->args[3].size()) // 能够成功转数字
            {
                stringObject *substr = val->substr(start_i, len);
                client->addReply(std::string(substr->c_str(), substr->_size) + shared.CRLF);
                substr->decrRefCount();
            }
            else
            {
                client->addReply("Can't convert to digit!\r\n");
            }
        }
    }

    key->decrRefCount();
}

static void incrCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);

    if (ret == dict.end())
    {
        client->addReply(shared.NotExist);
    }
    else
    {
        redisObject *obj = ret->second;
        if (obj->_type != TYPE_STRING)
        {
            client->addReply(shared.TypeMisMatch);
        }
        else
        {
            stringObject *str = (stringObject *)obj;
            str->incrVal(1);
            client->addReply(shared.OK);
            NotifyClientWatchedKeyChanged(client->_db_id, key);
        }
    }
    key->decrRefCount();
}

static void decrCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);

    if (ret == dict.end())
    {
        client->addReply(shared.NotExist);
    }
    else
    {
        redisObject *obj = ret->second;
        if (obj->_type != TYPE_STRING)
        {
            client->addReply(shared.TypeMisMatch);
        }
        else
        {
            stringObject *str = (stringObject *)obj;
            str->decrVal(1);
            client->addReply(shared.OK);
            NotifyClientWatchedKeyChanged(client->_db_id, key);
        }
    }
    key->decrRefCount();
}

static void mgetCommond(redisClient *client)
{
    int i = 1;
    while (i < client->args.size())
    {
        stringObject *key = new stringObject(client->args[i].c_str(), client->args[i].size());
        key->incrRefCount();

        auto &dict = redis_server._db[client->_db_id]._dict;
        auto ret = dict.find(key);
        if (ret == dict.end())
        {
            client->addReply(std::to_string(i) + ")" + shared.NotExist);
        }
        else
        {
            redisObject *obj = ret->second;
            if (obj->_type == TYPE_STRING)
            {
                stringObject *str = (stringObject *)obj;

                if (obj->_encoding == TYPE_Encoding_INT)
                    client->addReply(std::to_string(i) + ")" + std::to_string((long)str->c_str()) + shared.CRLF);
                else
                    client->addReply(std::to_string(i) + ")" + string(str->c_str(), str->_size) + shared.CRLF);
            }
            else
            {
                client->addReply(std::to_string(i) + ")" + shared.NotExist);
            }
        }

        key->decrRefCount();
        i++;
    }
}

static void rpushCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type != TYPE_LIST) // v不是链表对象
        {
            client->addReply(shared.TypeMisMatch);
        }
        else
        {
            listObject *listobj = (listObject *)obj;

            for (int i = 2; i < client->args.size(); i++)
                listobj->push_back(new stringObject(client->args[i].c_str(), client->args[i].size()));
            client->addReply(shared.OK);
            NotifyClientWatchedKeyChanged(client->_db_id, key);
        }
    }
    else // 没有链表对象
    {
        // 创建链表对象
        listObject *listobj = new listObject();

        for (int i = 2; i < client->args.size(); i++)
            listobj->push_back(new stringObject(client->args[i].c_str(), client->args[i].size()));

        dict.insert(make_pair(key, (redisObject *)listobj));
        client->addReply(shared.OK);
    }
    key->decrRefCount();
}

static void lpushCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type != TYPE_LIST) // v不是链表对象
        {
            client->addReply(shared.TypeMisMatch);
        }
        else
        {
            listObject *listobj = (listObject *)obj;

            for (int i = 2; i < client->args.size(); i++)
                listobj->push_front(new stringObject(client->args[i].c_str(), client->args[i].size()));
            client->addReply(shared.OK);
            NotifyClientWatchedKeyChanged(client->_db_id, key);
        }
    }
    else // 没有链表对象
    {
        // 创建链表对象
        listObject *listobj = new listObject();

        for (int i = 2; i < client->args.size(); i++)
            listobj->push_front(new stringObject(client->args[i].c_str(), client->args[i].size()));

        dict.insert(make_pair(key, (redisObject *)listobj));
        client->addReply(shared.OK);
    }
    key->decrRefCount();
}

static void rpopCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_LIST)
        {
            listObject *listobj = (listObject *)obj;
            listobj->pop_back();
            // 当链表为空时，删除key
            if (listobj->size() == 0)
            {
                delete listobj;
                dict.erase(key);
            }
            NotifyClientWatchedKeyChanged(client->_db_id, key);
            client->addReply(shared.OK);
        }
        else
        {
            client->addReply(shared.TypeMisMatch);
        }
    }
    else
    {
        client->addReply(shared.NotExist);
    }

    key->decrRefCount();
}

static void lpopCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_LIST)
        {
            listObject *listobj = (listObject *)obj;
            listobj->pop_front();
            // 当链表为空时，删除key
            if (listobj->size() == 0)
            {
                delete listobj;
                dict.erase(key);
            }
            NotifyClientWatchedKeyChanged(client->_db_id, key);
            client->addReply(shared.OK);
        }
        else
        {
            client->addReply(shared.TypeMisMatch);
        }
    }
    else
    {
        client->addReply(shared.NotExist);
    }

    key->decrRefCount();
}

static void lrangeCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        char *endptr1 = nullptr, *endptr2 = nullptr;
        int start = strtol(client->args[2].c_str(), &endptr1, 10);
        int end = strtol(client->args[3].c_str(), &endptr2, 10);
        if (client->args[2].size() == endptr1 - client->args[2].c_str() && client->args[3].size() == endptr2 - client->args[3].c_str())
        {
            listObject::iterator left;
            listObject::iterator right;

            if (start >= 0)
            {
                left = ((listObject *)ret->second)->begin();
                while (start--)
                {
                    left++;
                }
            }
            else
            {
                start = -start;
                left = ((listObject *)ret->second)->end();
                while (start--)
                {
                    left--;
                }
            }
            if (end >= 0)
            {
                right = ((listObject *)ret->second)->begin();
                while (end--)
                {
                    right++;
                }
            }
            else
            {
                end = -end;
                right = ((listObject *)ret->second)->end();
                while (end--)
                {
                    right--;
                }
            }
            int i = 1;
            while (true)
            {
                if (left == ((listObject *)ret->second)->end())
                {
                    left++;
                    continue;
                }

                if ((*left)->_encoding == TYPE_Encoding_SDS)
                {
                    client->addReply(std::to_string(i++) + ")" + string((*left)->c_str(), (*left)->_size) + "\r\n");
                }
                else
                {
                    client->addReply(std::to_string(i++) + ")" + std::to_string((long long)(*left)->c_str()) + "\r\n");
                }
                if (left == right)
                    break;

                left++;
            }
            client->addReply(shared.OK);
        }
        else // 区间转换失败
        {
            client->addReply("Can't convert to digit!\r\n");
        }
    }
    else
    {
        client->addReply(shared.NotExist);
    }
    key->decrRefCount();
}

static void llenCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_LIST)
        {
            listObject *listobj = (listObject *)obj;
            client->addReply(std::to_string(listobj->size()) + "\r\n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void lsetCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_LIST)
        {
            listObject *listobj = (listObject *)obj;
            char *endptr = nullptr;
            int index = strtol(client->args[2].c_str(), &endptr, 10);
            if (client->args[2].size() == endptr - client->args[2].c_str())
            {
                if (index > (int)listobj->size() || (-1) * index > (int)listobj->size()) // 避免整形提升
                    client->addReply("Subscript out of range!\r\n");
                else
                {
                    listObject::iterator iter;
                    if (index >= 0)
                    {
                        iter = listobj->begin();
                        while (index--)
                            iter++;
                    }
                    else
                    {
                        index = (-1) * index;
                        iter = listobj->end();
                        while (index--)
                            iter--;
                    }
                    (*iter)->decrRefCount();
                    (*iter) = nullptr;
                    (*iter) = new stringObject(client->args[3].c_str(), client->args[3].size());
                    (*iter)->incrRefCount();
                    client->addReply(shared.OK);
                }
            }
            else
                client->addReply("can't convert to digit!/r/n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void ltrimCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_LIST)
        {
            listObject *listobj = (listObject *)obj;
            char *endptr1 = nullptr;
            char *endptr2 = nullptr;
            int start = strtol(client->args[2].c_str(), &endptr1, 10);
            int end = strtol(client->args[3].c_str(), &endptr2, 10);
            if (client->args[2].size() == endptr1 - client->args[2].c_str() && client->args[3].size() == endptr2 - client->args[3].c_str())
            {
                listObject::iterator left;
                listObject::iterator right;
                if (start >= 0)
                {
                    left = listobj->begin();
                    while (start--)
                        left++;
                }
                else
                {
                    left = listobj->end();
                    start = (-1) * start;
                    while (start--)
                        left--;
                }
                if (end >= 0)
                {
                    right = listobj->begin();
                    while (end--)
                        right++;
                }
                else
                {
                    right = listobj->end();
                    end = (-1) * end;
                    while (end--)
                        right--;
                }
                left--;
                while (left != right)
                {
                    if (left == listobj->end())
                    {
                        left--;
                        continue;
                    }
                    listobj->erase(left--);
                }

                client->addReply(shared.OK);
            }
            else
                client->addReply("can't convert to digit!\r\n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void lremCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_LIST)
        {
            listObject *listobj = (listObject *)obj;
            stringObject *val = new stringObject(client->args[2].c_str(), client->args[2].size());
            val->incrRefCount();
            auto begin = listobj->begin();
            while (begin != listobj->end())
            {
                if (**begin == *val)
                    listobj->erase(begin++);
                else
                    begin++;
            }
            if (listobj->empty())
            {
                delete listobj;
                dict.erase(key);
            }
            val->decrRefCount();
            client->addReply(shared.OK);
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void rpoplpushCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_LIST)
        {
            listObject *listobj = (listObject *)obj;
            stringObject *str = *(--listobj->end());
            listobj->pop_back();
            listobj->push_front(str);
            client->addReply(shared.OK);
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void objectEcondingCommond(redisClient *client)
{
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto &dict = redis_server._db[client->_db_id]._dict;
    auto ret = dict.find(key);
    if (ret == dict.end())
    {
        client->addReply(shared.NotExist);
    }
    else
    {
        if (ret->second->_encoding == TYPE_Encoding_INT)
            client->addReply("digit \r\n");
        else if (ret->second->_encoding == TYPE_Encoding_SDS)
            client->addReply("SDS \r\n");
        else if(ret->second->_encoding == TYPE_Encoding_LIST)
            client->addReply("List\r\n");
        else if(ret->second->_type == TYPE_Ecoding_HASHTABLE)
            client->addReply("HashTable\r\n");
        else if(ret->second->_type == TYPE_Ecoding_INTSET)
            client->addReply("IntSet\r\n");
        else if(ret->second->_type == TYPE_Encoding_SKIPLIST)
            client->addReply("SkipList\r\n");
    }
    key->decrRefCount();
}

static void hsetCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_HASH)
        {
            if (client->args.size() % 2 != 0)
                client->addReply(shared.ArgumentNumError);
            else
            {
                int i = 2;
                while (i < client->args.size())
                {
                    stringObject *k = new stringObject(client->args[i].c_str(), client->args[i++].size());
                    stringObject *v = new stringObject(client->args[i].c_str(), client->args[i++].size());
                    HashObject *hashobj = (HashObject *)obj;
                    hashobj->replace({k, v});
                }
                client->addReply(shared.OK);
            }
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
    {
        if (client->args.size() % 2 != 0)
            client->addReply(shared.ArgumentNumError);
        else
        {
            HashObject *obj = new HashObject();
            int i = 2;
            while (i < client->args.size())
            {
                stringObject *k = new stringObject(client->args[i].c_str(), client->args[i++].size());
                stringObject *v = new stringObject(client->args[i].c_str(), client->args[i++].size());
                obj->insert({k, v});
            }
            dict.insert({key, obj});
            client->addReply(shared.OK);
        }
    }

    key->decrRefCount();
}

static void hsetnxCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_HASH)
        {
            if (client->args.size() % 2 != 0)
                client->addReply(shared.ArgumentNumError);
            else
            {
                int i = 2;
                while (i < client->args.size())
                {
                    stringObject *k = new stringObject(client->args[i].c_str(), client->args[i++].size());
                    stringObject *v = new stringObject(client->args[i].c_str(), client->args[i++].size());
                    HashObject *hashobj = (HashObject *)obj;
                    hashobj->insert({k, v});
                }
                client->addReply(shared.OK);
            }
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
    {
        if (client->args.size() % 2 != 0)
            client->addReply(shared.ArgumentNumError);
        else
        {
            HashObject *obj = new HashObject();
            int i = 2;
            while (i < client->args.size())
            {
                stringObject *k = new stringObject(client->args[i].c_str(), client->args[i++].size());
                stringObject *v = new stringObject(client->args[i].c_str(), client->args[i++].size());
                obj->insert({k, v});
            }
            dict.insert({key, obj});
            client->addReply(shared.OK);
        }
    }

    key->decrRefCount();
}

static void hgetallCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();

    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_HASH)
        {
            HashObject *hashobj = (HashObject *)obj;
            int i = 1;
            for (auto &e : *hashobj)
            {
                if (e.first->_encoding == TYPE_Encoding_SDS)
                    client->addReply(std::to_string(i) + ". { " + string(e.first->c_str(), e.first->_size));
                else
                    client->addReply(std::to_string(i) + ". { " + std::to_string((long long)e.first->c_str()));
                if (e.second->_encoding == TYPE_Encoding_SDS)
                    client->addReply(" , " + string(e.second->c_str(), e.second->_size) + " }" + "\r\n");
                else
                    client->addReply(" , " + std::to_string((long long)e.second->c_str()) + " }" + "\r\n");
                i++;
            }
            client->addReply(shared.OK);
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void hgetCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_HASH)
        {
            HashObject *hashobj = (HashObject *)obj;
            stringObject *k1 = new stringObject(client->args[2].c_str(), client->args[2].size());
            k1->incrRefCount();
            auto r1 = hashobj->find(k1);
            if (r1 != hashobj->end())
            {
                stringObject *v1 = r1->second;
                if (v1->_encoding == TYPE_Encoding_SDS)
                    client->addReply(string(v1->c_str(), v1->_size) + "\r\n");
                else
                    client->addReply(std::to_string((long long)v1->c_str()) + "\r\n");
            }
            else
                client->addReply("The Hashobj not exists this key!\r\n");

            k1->decrRefCount();
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void hdelCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_HASH)
        {
            HashObject *hashobj = (HashObject *)obj;
            stringObject *k1 = new stringObject(client->args[2].c_str(), client->args[2].size());
            k1->incrRefCount();
            hashobj->erase(k1);
            k1->decrRefCount();
            client->addReply(shared.OK);
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}

static void hlenCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_HASH)
        {
            HashObject *hashobj = (HashObject *)obj;
            client->addReply(std::to_string(hashobj->size()) + "\r\n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}

static void delCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        if (ret->second->_type == TYPE_STRING)
            delete (stringObject *)ret->second;
        else if (ret->second->_type == TYPE_LIST)
            delete (listObject *)ret->second;
        else if (ret->second->_type == TYPE_HASH)
            delete (HashObject *)ret->second;
        else if (ret->second->_type == TYPE_SET)
            delete (SetObject *)ret->second;
        else if (ret->second->_type == TYPE_ZSET)
            delete (ZsetObject *)ret->second;
        else
            assert(false);

        dict.erase(ret->first);
        client->addReply(shared.OK);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}
static void hkeysCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();

    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_HASH)
        {
            HashObject *hashobj = (HashObject *)obj;
            int i = 1;
            for (auto &e : *hashobj)
            {
                if (e.first->_encoding == TYPE_Encoding_SDS)
                    client->addReply(std::to_string(i) + ". " + string(e.first->c_str(), e.first->_size) + "\r\n");
                else
                    client->addReply(std::to_string(i) + ". " + std::to_string((long long)e.first->c_str()) + "\r\n");
                i++;
            }
            client->addReply(shared.OK);
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void hvalsCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();

    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_HASH)
        {
            HashObject *hashobj = (HashObject *)obj;
            int i = 1;
            for (auto &e : *hashobj)
            {
                if (e.second->_encoding == TYPE_Encoding_SDS)
                    client->addReply(std::to_string(i) + ". " + string(e.second->c_str(), e.second->_size) + "\r\n");
                else
                    client->addReply(std::to_string(i) + ". " + std::to_string((long long)e.second->c_str()) + "\r\n");
                i++;
            }
            client->addReply(shared.OK);
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void saddCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type != TYPE_SET)
            client->addReply(shared.TypeMisMatch);
        else
        {
            SetObject *setobj = (SetObject *)obj;
            for (int i = 2; i < client->args.size(); i++)
            {
                // cout<<"arg"<<i<<":"<<client->args[i]<<endl;
                stringObject *str = new stringObject(client->args[i].c_str(), client->args[i].size());
                setobj->add(str);
            }
            client->addReply(shared.OK);
        }
    }
    else
    {
        SetObject *setobj = new SetObject();
        for (int i = 2; i < client->args.size(); i++)
        {

            stringObject *str = new stringObject(client->args[i].c_str(), client->args[i].size());
            str->incrRefCount();
            setobj->add(str);
            str->decrRefCount();
        }
        dict.insert({key, setobj});
        client->addReply(shared.OK);
        /* cout<<"setobj size:"<<setobj->size()<<endl;
         cout<<"setobj type:"<<setobj->_type<<endl;
         cout<<"setobj encoding:"<<setobj->_encoding<<endl;*/
    }
    key->decrRefCount();
}

static void smembersCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_SET)
        {
            SetObject *setobj = (SetObject *)obj;
            if (setobj->size() == 0)
            {
                client->addReply("empty\r\n");
            }
            else
            {
                int i = 1;
                if (obj->_encoding == TYPE_Ecoding_HASHTABLE)
                {
                    unordered_set<stringObject *> *uset = (unordered_set<stringObject *> *)(setobj->_ptr);
                    for (auto strobj : *uset)
                    {
                        if (strobj->_encoding == TYPE_Encoding_SDS)
                            client->addReply(std::to_string(i) + ". " + std::string(strobj->c_str(), strobj->_size) + "\r\n");
                        else if (strobj->_encoding == TYPE_Encoding_INT)
                        {
                            client->addReply(std::to_string(i) + " " + std::to_string((long long)strobj->c_str()) + "\r\n");
                        }
                        else
                            assert(false);

                        i++;
                    }
                }
                else if (obj->_encoding == TYPE_Ecoding_INTSET)
                {
                    intset *pintset = (intset *)(setobj->_ptr);
                    for (int j = 0; j < pintset->_length; j++)
                    {
                        long long num = intset::getSizePtr(pintset->_data + j * pintset->_encoding, pintset->_encoding);
                        client->addReply(std::to_string(i) + ". " + std::to_string(num) + "\r\n");
                        i++;
                    }
                }
                else
                    assert(false);
            }
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}

static void scardCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_SET)
        {
            int size = ((SetObject *)obj)->size();
            client->addReply(std::to_string(size) + "\r\n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void sremCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);

    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_SET)
        {
            SetObject *setobj = (SetObject *)obj;
            for (int i = 2; i < client->args.size(); i++)
            {
                stringObject *v = new stringObject(client->args[i].c_str(), client->args[i].size());
                v->incrRefCount();
                setobj->erase(v);
                v->decrRefCount();
            }
            client->addReply(shared.OK);
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void smoveCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *v = new stringObject(client->args[1].c_str(), client->args[1].size());
    stringObject *srckey = new stringObject(client->args[2].c_str(), client->args[2].size());
    stringObject *dstkey = new stringObject(client->args[3].c_str(), client->args[3].size());
    v->incrRefCount();
    srckey->incrRefCount();
    dstkey->incrRefCount();
    auto ret1 = dict.find(srckey);
    auto ret2 = dict.find(dstkey);

    if (ret1 == dict.end())
        client->addReply("src not exists\r\n");
    else if (ret2 == dict.end())
        client->addReply("dst not exists\r\n");
    else
    {
        SetObject *srcobj = (SetObject *)(ret1->second);
        SetObject *dstobj = (SetObject *)(ret2->second);
        if (srcobj->find(v))
        {
            srcobj->erase(v);
            dstobj->add(v);
            client->addReply(shared.OK);
        }
        else
            client->addReply("val not exists\r\n");
    }
    v->decrRefCount();
    srckey->decrRefCount();
    srckey->decrRefCount();
}

static void sinterCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    vector<SetObject *> setobjs(client->args.size() - 1, nullptr);
    for (int i = 1; i < client->args.size(); i++)
    {
        stringObject *key = new stringObject(client->args[i].c_str(), client->args[i].size());
        key->incrRefCount();
        auto ret = dict.find(key);
        if (ret != dict.end())
        {
            redisObject *obj = ret->second;
            if (obj->_type == TYPE_SET)
            {
                setobjs[i - 1] = (SetObject *)obj;
            }
            else
            {
                client->addReply(std::to_string(i) + ". " + shared.TypeMisMatch);
                key->decrRefCount();
                return;
            }
        }
        else
        {
            client->addReply(std::to_string(i) + ". " + shared.NotExist);
            key->decrRefCount();
            return;
        }

        key->decrRefCount();
    }

    // 到这里，说明集合都存在且类型正确
    SetObject *setobj = setobjs[0];
    std::list<stringObject *> _list;

    if (setobj->_encoding == TYPE_Ecoding_HASHTABLE)
    {
        // auto begin = ((unordered_set<stringObject*>*)setobj->_ptr)->begin();
        for (auto &str : *((unordered_set<stringObject *> *)setobj->_ptr))
        {
            int i;
            for (i = 1; i < setobjs.size(); i++)
            {
                if (setobjs[i]->find(str) == false)
                    break;
            }
            if (i == setobjs.size())
                _list.push_back(str);
        }
    }
    else
    {
        // 如果为TYPE_Ecoing_INSET 先不写了
    }

    int i = 1;
    for (auto &strobj : _list)
    {
        if (strobj->_encoding == TYPE_Encoding_INT)
        {
            client->addReply(std::to_string(i) + ". " + std::to_string((long long)strobj->c_str()) + "\r\n");
        }
        else if (strobj->_encoding == TYPE_Encoding_SDS)
        {
            client->addReply(std::to_string(i) + ". " + string(strobj->c_str(), strobj->_size) + "\r\n");
        }
        else
            assert(false);
        i++;
    }
}

static void sunionCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    vector<SetObject *> setobjs(client->args.size() - 1, nullptr);
    for (int i = 1; i < client->args.size(); i++)
    {
        stringObject *key = new stringObject(client->args[i].c_str(), client->args[i].size());
        key->incrRefCount();
        auto ret = dict.find(key);
        if (ret != dict.end())
        {
            redisObject *obj = ret->second;
            if (obj->_type == TYPE_SET)
            {
                setobjs[i - 1] = (SetObject *)obj;
            }
            else
            {
                client->addReply(std::to_string(i) + ". " + shared.TypeMisMatch);
                key->decrRefCount();
                return;
            }
        }
        else
        {
            client->addReply(std::to_string(i) + ". " + shared.NotExist);
            key->decrRefCount();
            return;
        }

        key->decrRefCount();
    }

    // 到这里，说明集合都存在且类型正确
    SetObject *setobj = new SetObject();
    for (int i = 0; i < setobjs.size(); i++)
    {
        for (auto &strobj : *((unordered_set<stringObject *> *)setobjs[i]->_ptr))
        {
            setobj->add(strobj);
        }
    }
    int i = 1;
    for (auto &strobj : *((unordered_set<stringObject *> *)setobj->_ptr))
    {
        if (strobj->_encoding == TYPE_Encoding_INT)
        {
            client->addReply(std::to_string(i) + ". " + std::to_string((long long)strobj->c_str()) + "\r\n");
        }
        else if (strobj->_encoding == TYPE_Encoding_SDS)
        {
            client->addReply(std::to_string(i) + ". " + string(strobj->c_str(), strobj->_size) + "\r\n");
        }
        else
            assert(false);
        i++;
    }
}

static void zaddCommond(redisClient *client)
{
    if (client->args.size() % 2 != 0)
    {
        client->addReply(shared.ArgumentNumError);
        return;
    }

    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);

    if (ret == dict.end())
    {
        ZsetObject *zsetobj = new ZsetObject();
        dict.insert(make_pair(key, (redisObject *)zsetobj));
    }
    else
    {
        if (ret->second->_type != TYPE_ZSET)
        {
            cout << ret->second->_type << endl;
            client->addReply(shared.TypeMisMatch);
            return;
        }
    }

    ret = dict.find(key);

    ZsetObject *zsetobj = (ZsetObject *)(ret->second);
    for (int i = 2; i < client->args.size(); i++)
    {
        char *endptr = nullptr;
        int score = strtol(client->args[i].c_str(), &endptr, 10);
        if (endptr && *endptr == '\0')
        {
            stringObject *str = new stringObject(client->args[i + 1].c_str(), client->args[i + 1].size());
            str->incrRefCount();
            zsetobj->insert(score, str);
            str->decrRefCount();
        }
        i++;
    }
    client->addReply(shared.OK);
    key->decrRefCount();
}

static void zrangeCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        if (ret->second->_type == TYPE_ZSET)
        {
            char *endptr1 = nullptr;
            char *endptr2 = nullptr;
            int leftIndex = strtol(client->args[2].c_str(), &endptr1, 10);
            int rightIndex = strtol(client->args[3].c_str(), &endptr2, 10);
            if (endptr1 && endptr2 && *endptr1 == '\0' && *endptr2 == '\0')
            {
                ZsetObject *zsetobj = (ZsetObject *)(ret->second);
                auto range = zsetobj->zrange(leftIndex, rightIndex);
                if (range.empty())
                    client->addReply("empty\r\n");
                else
                {
                    for (int i = 0; i < range.size(); i++)
                    {
                        client->addReply(std::to_string(i + 1) + ". { " + std::to_string(range[i].first));
                        client->addReply(" , " + string(range[i].second->c_str(), range[i].second->_size) + " } \r\n");
                    }
                }
            }
            else
                client->addReply("conver to num fail!\r\n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void zcountCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *redisobj = ret->second;
        if (redisobj->_type == TYPE_ZSET)
        {
            ZsetObject *zsetobj = (ZsetObject *)redisobj;

            char *endptr1 = nullptr;
            char *endptr2 = nullptr;
            int leftIndex = strtol(client->args[2].c_str(), &endptr1, 10);
            int rightIndex = strtol(client->args[3].c_str(), &endptr2, 10);
            if (endptr1 && endptr2 && *endptr1 == '\0' && *endptr2 == '\0')
            {
                client->addReply(std::to_string(zsetobj->zcount(leftIndex, rightIndex)) + "\r\n");
            }
            else
                client->addReply("convert to dight fail!\r\n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}

static void zincrbyCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        redisObject *obj = ret->second;
        if (obj->_type == TYPE_ZSET)
        {
            ZsetObject *zsetobj = (ZsetObject *)obj;
            char *endptr = nullptr;
            int increment = strtol(client->args[3].c_str(), &endptr, 10);
            if (endptr && *endptr == '\0')
            {
                stringObject *str = new stringObject(client->args[2].c_str(), client->args[2].size());
                str->incrRefCount();

                bool r = zsetobj->zincrby(str, increment);
                if (r == true)
                    client->addReply(shared.OK);
                else
                    client->addReply("member not exists!\r\n");

                str->decrRefCount();
            }
            else
                client->addReply("convert to num fail!\r\n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}

static void zscoreCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        if (ret->second->_type == TYPE_ZSET)
        {
            ZsetObject *zsetobj = (ZsetObject *)(ret->second);
            stringObject *k = new stringObject(client->args[2].c_str(), client->args[2].size());
            k->incrRefCount();
            auto rt = zsetobj->zscore(k);
            if (rt.second == true)
                client->addReply(std::to_string(rt.first) + "\r\n");
            else
                client->addReply("member not exists!\r\n");
            k->decrRefCount();
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}

static void zrangeByscoreCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        if (ret->second->_type == TYPE_ZSET)
        {
            char *endptr1 = nullptr;
            char *endptr2 = nullptr;
            int leftscore = strtol(client->args[2].c_str(), &endptr1, 10);
            int rightscore = strtol(client->args[3].c_str(), &endptr2, 10);
            if (endptr1 && endptr2 && *endptr1 == '\0' && *endptr2 == '\0')
            {
                ZsetObject *zsetobj = (ZsetObject *)(ret->second);
                auto con = zsetobj->zrangeByscore(leftscore, rightscore);
                if (!con.empty())
                {
                    for (int i = 0; i < con.size(); i++)
                    {
                        client->addReply(std::to_string(i + 1) + ".{ " + std::to_string(con[i].first) + " , ");
                        client->addReply(string(con[i].second->c_str(), con[i].second->_size) + " }\r\n");
                    }
                }
                else
                    client->addReply("empty\r\n");
            }
            else
                client->addReply("convert to num fail!\r\n");
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);

    key->decrRefCount();
}

static void zrankCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        if (ret->second->_type == TYPE_ZSET)
        {
            ZsetObject *zsetobj = (ZsetObject *)ret->second;
            stringObject *str = new stringObject(client->args[2].c_str(), client->args[2].size());
            str->incrRefCount();
            int index = zsetobj->zrank(str);
            if (index != -1)
                client->addReply(std::to_string(index) + "\r\n");
            else
                client->addReply("member not exists!\r\n");
            str->decrRefCount();
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}

static void zremCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    stringObject *key = new stringObject(client->args[1].c_str(), client->args[1].size());
    key->incrRefCount();
    auto ret = dict.find(key);
    if (ret != dict.end())
    {
        if (ret->second->_type == TYPE_ZSET)
        {
            ZsetObject *zsetobj = (ZsetObject *)ret->second;
            vector<stringObject *> remobjs;
            for (int i = 2; i < client->args.size(); i++)
            {
                stringObject *k = new stringObject(client->args[i].c_str(), client->args[i].size());
                k->incrRefCount();
                remobjs.push_back(k);
            }
            zsetobj->zrem(remobjs);
            for (auto *strobj : remobjs)
                strobj->decrRefCount();

            client->addReply(shared.OK);
        }
        else
            client->addReply(shared.TypeMisMatch);
    }
    else
        client->addReply(shared.NotExist);
    key->decrRefCount();
}

static void sdiffCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;
    vector<SetObject *> setobjs(client->args.size() - 1, nullptr);
    for (int i = 1; i < client->args.size(); i++)
    {
        stringObject *key = new stringObject(client->args[i].c_str(), client->args[i].size());
        key->incrRefCount();
        auto ret = dict.find(key);
        if (ret != dict.end())
        {
            redisObject *obj = ret->second;
            if (obj->_type == TYPE_SET)
            {
                setobjs[i - 1] = (SetObject *)obj;
            }
            else
            {
                client->addReply(std::to_string(i) + ". " + shared.TypeMisMatch);
                key->decrRefCount();
                return;
            }
        }
        else
        {
            client->addReply(std::to_string(i) + ". " + shared.NotExist);
            key->decrRefCount();
            return;
        }

        key->decrRefCount();
    }

    // 到这里，说明集合都存在且类型正确
    SetObject *setobj = setobjs[0];
    std::list<stringObject *> _list;

    if (setobj->_encoding == TYPE_Ecoding_HASHTABLE)
    {
        for (auto &str : *((unordered_set<stringObject *> *)setobj->_ptr))
        {
            int i;
            for (i = 1; i < setobjs.size(); i++)
            {
                if (setobjs[i]->find(str) == true)
                    break;
            }
            if (i == setobjs.size())
                _list.push_back(str);
        }
    }
    else
    {
        // 如果为TYPE_Ecoing_INSET 先不写了
    }

    int i = 1;
    for (auto &strobj : _list)
    {
        if (strobj->_encoding == TYPE_Encoding_INT)
        {
            client->addReply(std::to_string(i) + ". " + std::to_string((long long)strobj->c_str()) + "\r\n");
        }
        else if (strobj->_encoding == TYPE_Encoding_SDS)
        {
            client->addReply(std::to_string(i) + ". " + string(strobj->c_str(), strobj->_size) + "\r\n");
        }
        else
            assert(false);
        i++;
    }
}

static void existsCommond(redisClient *client)
{
    auto &dict = redis_server._db[client->_db_id]._dict;

    stringObject *obj = new stringObject(client->args[1].c_str(), client->args[1].size());
    obj->incrRefCount();

    if (dict.find(obj) == dict.end())
    {
        client->addReply(shared.NotExist);
    }
    else
    {
        client->addReply("Exists\r\n");
    }

    obj->decrRefCount();
}

static void selectCommond(redisClient *client)
{
    char *endptr;
    int db_id = strtol(client->args[1].c_str(), &endptr, 10);

    if (endptr - client->args[1].c_str() == client->args[1].size())
    {
        if (REDIS_OK == selectDb(client, db_id))
            client->addReply(shared.OK);
        else
            client->addReply("The legal range of numbers is 0 to 15 !\r\n");
    }
    else
        client->addReply("Can't translate to num!\r\n");
}

static void multiCommond(redisClient *client)
{
    client->flag |= MUILTI;
    client->addReply(shared.OK);
}

static void execCommond(redisClient *client)
{
    auto cleanWatchKey = [&]()
    {
        for (auto o : client->_watchkeys)
        {
            auto ret = redis_server._db->_watchkeys.find(o);
            auto &watch_client_lists = ret->second;
            auto iter = watch_client_lists.begin();
            while (iter != watch_client_lists.end())
            {
                if (*iter == client)
                {
                    watch_client_lists.erase(iter);
                    break;
                }
                iter++;
            }
        }
        client->_watchkeys.clear();
    };
    if (!(client->flag & MUILTI))
    {
        client->addReply("Transaction not started\r\n");
        return;
    }

    if (client->flag & DIRTY_DAS)
    {
        client->addReply("watchkeys changed\r\n");
        client->flag &= ~MUILTI;
        client->multiQueue.clear();
        cleanWatchKey();
        client->flag &= ~DIRTY_DAS;
        return;
    }

    if (client->multiQueue.empty())
    {
        client->addReply("empty commond queue\r\n");
        client->flag &= ~MUILTI;
        cleanWatchKey();
        return;
    }

    int i = 1;
    while (!client->multiQueue.empty())
    {
        client->args = client->multiQueue.front().args;

        client->addReply(std::to_string(i++) + ".");

        call(client, client->multiQueue.front().cmd);

        client->multiQueue.pop_front();
    }
    client->args.clear();

    cleanWatchKey();
    client->flag &= ~MUILTI;
    client->flag &= ~DIRTY_DAS;
}

static void watchCommond(redisClient *client)
{
    auto &dict = redis_server._db->_dict;
    stringObject *o1 = new stringObject(client->args[1].c_str(), client->args[1].size());
    o1->incrRefCount();

    for (auto watchkey : client->_watchkeys)
    {
        if (*watchkey == *o1)
        {
            o1->decrRefCount();
            client->addReply("Already watch this key\r\n");
            return;
        }
    }

    // key不存在
    if (dict.end() == dict.find(o1))
    {
        client->addReply("watchkey is not existed\r\n");
    }
    else // key存在
    {
        auto &watchkey = redis_server._db->_watchkeys;
        auto ret = watchkey.find(o1);
        if (watchkey.end() == ret) // 没有监视key的客户端
        {
            std::list<redisClient *> l;
            l.push_back(client);
            watchkey.insert({o1, l});
        }
        else
        {
            ret->second.push_back(client);
        }

        client->addReply(shared.OK);
    }
    o1->decrRefCount();
}

static void pexpireatCommond(redisClient *client)
{
    auto &dict = redis_server._db->_dict;
    // 查看key是否存在
    stringObject *o1 = new stringObject(client->args[1].c_str(), client->args[1].size());
    o1->incrRefCount();
    if (dict.find(o1) == dict.end())
    {
        client->addReply("key not existed!\r\n");
        return;
    }

    long now_sec, now_ms;
    aeGetTime(now_sec, now_ms);
    long long now = now_sec * 1000 + now_ms; // 当前时间，精度为毫秒

    char *endptr = nullptr;
    long long expiretime = strtoll(client->args[2].c_str(), &endptr, 10);
    cout << "client->args.size()" << client->args.size() << endl;
    cout << "expiretime" << expiretime << endl;
    if (endptr && endptr - client->args[2].c_str() == client->args[2].size())
    { // 转成功
        if (now < expiretime)
        {
            cout << "设置过期时间成功" << endl;
            auto &expire = redis_server._db->_expire;
            expire.replace({o1, expiretime});
        }
        else
        {
            client->addReply("time error\r\n");
        }
        client->addReply("now(毫秒):    " + std::to_string(now) + string("\r\n"));
    }
    else
    {
        client->addReply("time is not num\r\n");
    }

    o1->decrRefCount();
}

static void subscribeCommond(redisClient* client)
{
    auto ret = redis_server._pubsub_channels.find(client->args[1]);

    if(ret == redis_server._pubsub_channels.end())
        redis_server._pubsub_channels.insert({client->args[1],std::list<redisClient*>()});

    redis_server._pubsub_channels[client->args[1]].push_back(client);
    
    client->addReply(shared.OK);
}

static void unsubscribeCommond(redisClient* client)
{
    auto ret = redis_server._pubsub_channels.find(client->args[1]);
    if(ret == redis_server._pubsub_channels.end())
    {
        client->addReply("chnneal not exists!\r\n");
    }
    else
    {
        auto& sub_clients = ret->second;
        auto begin = sub_clients.begin();
        bool flag=false;
        while(begin!=sub_clients.end())
        {
            if(*begin == client)
            {
                sub_clients.erase(begin);
                flag=true;
                break;
            }
        }
        if(flag)
        client->addReply(shared.OK);
        else
        client->addReply("you hadn't subscribe the channel!\r\n");
    }
}

static void publishCommond(redisClient* client)
{
    auto& channel = client->args[1];
    auto ret = redis_server._pubsub_channels.find(channel);
    if(ret != redis_server._pubsub_channels.end())
    {
       auto& channel_clients = ret->second;
       
       for(auto* cli : channel_clients)
       cli->addReply(client->args[2]);
    }
    else
    client->addReply("the channel not exists!\r\n");
}

static void psubscribeCommond(redisClient* client)
{
   redis_server.pubsub_patterns.push_back({client->args[1],client});
   client->addReply(shared.OK);
}
static void punsubscribeCommond(redisClient* client)
{
   bool flag = false;
   auto begin = redis_server.pubsub_patterns.begin();
   while(begin!=redis_server.pubsub_patterns.end())
   {
     if(begin->pattern == client->args[1] && begin->client == client)
     {
        flag=true;
        redis_server.pubsub_patterns.erase(begin);
        break;
     }
   }
   if(flag)
   client->addReply(shared.OK);
   else
   client->addReply("hadn't subscribe the pattern!\r\n");
}

static void bgrewriteaofCommond(redisClient* )
{
    if(redis_server.state & BGREWRITEAOF) //已经服务器在重写了
    return;

    //开始重写
    redis_server.state &= BGREWRITEAOF;
    //创建子进程来重写
    pid_t pid = fork();
    if(pid == -1)
    {
        cerr<<"fork fail !"<<endl;
        exit(-1);
    }
    if(pid!=0)
    {//父进程
       redis_server.bgrewriteaofPid=pid;
    }
    else
    {//子进程
        
        bgrewriteaof();
        exit(0);
    }
}


void aeMain(aeEvents *aeEv)
{
    while (1)
    {
        long sec = aeEv->getNearestTime();
        timeval tv = {sec, 0};

        aeEv->getFiledEvents(&tv);
        aeEv->processFiledEvents();

        aeEv->processTimeEvents();
    }
}

int main()
{
    cout << "redis_server sockfd:" << redis_server._fd << endl;
    aeMain(&redis_server._aeEv);

    return 0;
}
